package com.shujia.flink.window

import com.shujia.flink.window.Demo4TImeWindow.Event
import org.apache.flink.api.common.functions.{RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.state.ListStateDescriptor
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.apache.flink.api.common.state.ListState
import org.apache.flink.streaming.api.functions.KeyedProcessFunction

import scala.collection.mutable.ListBuffer

object Demo4TImeWindow {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val ds = env.socketTextStream("node1", 9999)

    /**
      * 统计最近20秒最热门的前2个商品   5秒统计一次
      *
      * uid,itemId,action
      * 1,112,1
      * 2,112,1
      * 3,112,1
      * 4,112,1
      * 5,112,1
      * 1,113,1
      * 2,113,1
      * 3,113,1
      * 4,113,1
      * 1,114,1
      * 2,115,1
      * 3,116,1
      *
      */

    ds
      .map(_.split(","))
      .map(arr => (arr(1), 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .process(new MyProcessWindowFunction)
      .keyBy(_._3) //以窗口时间进行分组
      .process(new MyKeyedProcessFunction)
      .print()


    env.execute("Demo1Window")

  }

  case class Event(itemDd: String, c: Int)


}


class MyKeyedProcessFunction extends KeyedProcessFunction[Long, (String, Int, Long, Long), (String, Long)] {

  //存一个窗口所有数据的状态    一个key的状态
  var topNState: ListState[Event] = _

  override def open(parameters: Configuration): Unit = {

    //初始化状态
    val listDesc = new ListStateDescriptor[Event]("topN", classOf[Event])
    topNState = getRuntimeContext.getListState(listDesc)
  }

  //每一条数据都会执行一次
  override def processElement(value: (String, Int, Long, Long), ctx: KeyedProcessFunction[Long, (String, Int, Long, Long), (String, Long)]#Context, out: Collector[(String, Long)]): Unit = {

    //将当前商品保存到状态中
    topNState.add(Event(value._1, value._2))

    //当窗口所有数据到达到之后触发   注册窗口结束时间为定时器
    ctx.timerService().registerProcessingTimeTimer(value._4 + 1)

  }

  //当定时器时间到了之后触发执行
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, (String, Int, Long, Long), (String, Long)]#OnTimerContext, out: Collector[(String, Long)]): Unit = {


    val topList = new ListBuffer[Event]

    val iter = topNState.get().iterator()

    while (iter.hasNext) {
      topList.+=(iter.next())
    }


    //取topn  返回结果
    topList
      .sortBy(event => -event.c) //商品点击数量倒序排序
      .take(2)
      .foreach(event => {
        //将数据发生到下游
        out.collect(event.itemDd, event.c)
      })
  }
}


//统计同一个窗口中同一个商品的数量
class MyProcessWindowFunction extends ProcessWindowFunction[(String, Int), (String, Int, Long, Long), String, TimeWindow] {
  override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[(String, Int, Long, Long)]): Unit = {
    //窗口结束时间
    val end = context.window.getEnd
    //窗口开始时间
    val start = context.window.getStart

    var sum = 0

    //统计窗口内商品出现的次数
    elements.foreach(kv => sum += 1)

    //返回结果
    out.collect((key, sum, start, end))
  }
}
